package com.bw.data_summary;

import org.apache.flink.runtime.state.storage.FileSystemCheckpointStorage;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.api.TableResult;
//  门禁类型
public class dwd_access_control_entrance {
    public static void main(String[] args) throws Exception {
        //1.获取Stream的执行环境
        StreamExecutionEnvironment senv = StreamExecutionEnvironment.getExecutionEnvironment();
        senv.setParallelism(1);

        //2.创建表的执行环境
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(senv);


        //3.读取门禁刷卡事实表ods_entrance_guard_log
        tEnv.executeSql("CREATE TABLE `bean_entrance_guard_log`(\n" +
                " id STRING,\n" +
                " openMode SMALLINT,\n" +
                " openTime STRING,\n" +
                " ownerId INT,\n" +
                " proc_time as PROCTIME()\n" +
                ") WITH (\n" +
                " 'connector' = 'kafka',\n" +
                " 'topic' = 'entrance_guard_log',\n" +
                " 'properties.bootstrap.servers' = 'node102:9092',\n" +
                " 'properties.group.id' = 'entrance_guard_log',\n" +
                " 'scan.startup.mode' = 'earliest-offset',\n" +
                " 'format' = 'json'\n" +
                ")\n");

        //4.业主维表
        tEnv.executeSql("CREATE TABLE dim_base_ower(\n" +
                " id INT,\n" +
                " roomId INT,\n" +
                " owerCode STRING,\n" +
                " owerName STRING,\n" +
                " owerSex STRING,\n" +
                " owerAge INT,\n" +
                " owerIdcard STRING,\n" +
                " owerTel STRING,\n" +
                " owerHomeplace STRING,\n" +
                " owerWorkplace STRING,\n" +
                " remark STRING,\n" +
                " createTime STRING,\n" +
                " operatorTime STRING,\n" +
                " PRIMARY KEY (id) NOT ENFORCED\n" +
                ")  WITH (\n" +
                "    'connector' = 'doris',\n" +
                "    'fenodes' = 'node102:8030',\n" +
                "    'table.identifier' = 'sca_doris.ower',\n" +
                "    'username' = 'root',\n" +
                "    'password' = '123456',\n" +
                "    'jdbc-url' = 'jdbc:mysql://node102:9030?useSSL=false',\n" +  // 添加SSL配置
                "    'lookup.jdbc.async' = 'true'\n" +
                ")\n");

        //5. 房间维度表
        tEnv.executeSql("CREATE TABLE dim_base_room(\n" +
                " id INT,\n" +
                " houseId INT,\n" +
                " roomCode STRING,\n" +
                " roomName STRING,\n" +
                " peopleNum INT,\n" +
                " remark STRING,\n" +
                " PRIMARY KEY (id) NOT ENFORCED\n" +
                ") WITH (\n" +
                "    'connector' = 'doris',\n" +
                "    'fenodes' = 'node102:8030',\n" +
                " 'table.identifier' = 'sca_doris.room',\n" +
                "    'username' = 'root',\n" +
                "    'password' = '123456',\n" +
                "    'jdbc-url' = 'jdbc:mysql://node102:9030?useSSL=false',\n" +  // 添加SSL配置
                "    'lookup.jdbc.async' = 'true'\n" +
                ")\n");


        //6. 楼栋维度表
        tEnv.executeSql("CREATE TABLE dim_base_house(\n" +
                " id INT,\n" +
                " cid INT,\n" +
                " houseCode STRING,\n" +
                " houseName STRING,\n" +
                " buildStartTime STRING,\n" +
                " buildEndTime STRING,\n" +
                " houseArea FLOAT,\n" +
                " remark STRING,\n" +
                " PRIMARY KEY (id) NOT ENFORCED\n" +
                ")  WITH (\n" +
                "    'connector' = 'doris',\n" +
                "    'fenodes' = 'node102:8030',\n" +
                "    'table.identifier' = 'sca_doris.house',\n" +
                "    'username' = 'root',\n" +
                "    'password' = '123456',\n" +
                "    'jdbc-url' = 'jdbc:mysql://node102:9030?useSSL=false',\n" +  // 添加SSL配置
                "    'lookup.jdbc.async' = 'true'\n" +
                ")\n");

        //7. 小区维度表
        tEnv.executeSql("CREATE TABLE dim_base_community(\n" +
                " id INT,\n" +
                " communityName STRING,\n" +
                " district_id INT,\n" +
                " city_id INT,\n" +
                " communityLocation STRING,\n" +
                " remark STRING,\n" +
                " PRIMARY KEY (id) NOT ENFORCED\n" +
                ") WITH (\n" +
                "    'connector' = 'doris',\n" +
                "    'fenodes' = 'node102:8030',\n" +
                "    'table.identifier' = 'sca_doris.community',\n" +
                "    'username' = 'root',\n" +
                "    'password' = '123456',\n" +
                "    'jdbc-url' = 'jdbc:mysql://node102:9030?useSSL=false',\n" +  // 添加SSL配置
                "    'lookup.jdbc.async' = 'true'\n" +
                ")\n");

        //8.事实表与维度表关联
        Table resultTable = tEnv.sqlQuery("select " +
                "b1.id, " +
                "b1.ownerId, " +
                "b1.openTime, " +
                "d3.cid as cId, " +
                "d3.houseCode, "+
                "d2.roomCode, "+
                "b1.openMode " +
                "from bean_entrance_guard_log b1 " +
                "left join dim_base_ower for system_time as of b1.proc_time as d1 " +
                "on b1.ownerId=d1.id "+
                "left join dim_base_room for system_time as of b1.proc_time as  d2 "+
                "on d1.roomId=d2.id "+
                "left join dim_base_house for system_time as of b1.proc_time as d3 "+
                "on d2.houseId=d3.id "+
                "left join dim_base_community for system_time as of b1.proc_time as d4 "+
                "on d3.cid=d4.id"
        );
        tEnv.createTemporaryView("resultTable",resultTable);

        //9.创建dwd_entrance_guard_log表
        tEnv.executeSql("CREATE TABLE dwd_entrance_guard_log (\n" +
                "id STRING,\n" +
                "ownerId INT,\n" +
                "openTime STRING,\n" +
                "cId INT,\n" +
                "houseCode STRING,\n" +
                "roomCode STRING,\n" +
                "openMode SMALLINT\n" +
                ") WITH (\n" +
                " 'connector' = 'kafka',\n" +
                " 'topic' = 'dwd_entrance_guard_log',\n" +
                " 'properties.bootstrap.servers' = 'node101:9092',\n" +
                " 'properties.group.id' = 'dwd_entrance_guard_log_group',\n" +
                " 'format' = 'json',\n" +
                " 'scan.startup.mode' = 'earliest-offset'\n" +
                ")\n");

        //10.flink sql 将查询结果写入dwd_entrance_guard_log
        TableResult tableResult = tEnv.executeSql("INSERT INTO dwd_entrance_guard_log SELECT * FROM resultTable");

        // 可选：打印作业执行结果
        // tableResult.print();

        // 提交Flink作业
        senv.execute("DwdEntranceGuardLog");
    }

}
